package com.cetc.cloud.kmga.util;

import com.rabbitmq.client.*;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import java.io.IOException;
import java.io.InputStream;
import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;

/**
 * Created by Administrator on 2017/9/12 0012.
 */
public class MqUtil {

    private static Log LOGGER = LogFactory.getLog(MqUtil.class);
    volatile private static MqUtil instance;

    private static String MQ_HOST;
    private static String MQ_USER;
    private static String MQ_PWD;
    private static String MQ_NODE_LIST;

    private static Properties MQ_PRO;
    private final static String MQ_URL = "conf/MqService.properties";

    private static Connection connection;
    private static Channel channel;


    static {
        MQ_PRO = new Properties();
        try {
            InputStream inputStream = MqUtil.class.getClassLoader().getResourceAsStream(MQ_URL);
            MQ_PRO.load(inputStream);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public MqUtil(String MQ_HOST, String MQ_USER, String MQ_PWD) {
        this.MQ_HOST = MQ_HOST;
        this.MQ_USER = MQ_USER;
        this.MQ_PWD = MQ_PWD;
        this.MQ_NODE_LIST = MQ_PRO.getProperty("MQ_NODE_LIST");
        init();
    }


    public static MqUtil getInstance(String MQ_HOST, String MQ_USER, String MQ_PWD) {
        try {
            if (instance != null) {//懒汉式

            } else {
                //创建实例之前可能会有一些准备性的耗时工作
                Thread.sleep(300);
                synchronized (MqUtil.class) {
                    if (instance == null) {//二次检查
                        instance = new MqUtil(MQ_HOST, MQ_USER, MQ_PWD);

                    }
                }
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
            instance = null;
        }
        return instance;
    }

    private static void init() {
        try {
            ConnectionFactory factory = new ConnectionFactory();
            //设置MabbitMQ所在主机ip或者主机名
//            factory.setHost(MQ_HOST);
            factory.setUsername(MQ_USER);
            factory.setPassword(MQ_PWD);
            if (StringUtils.isNotBlank(MQ_NODE_LIST)) {
                List<Address> addressList = new ArrayList<>();
                String[] nodeArr = MQ_NODE_LIST.split(",");
                for (int i = 0; i < nodeArr.length; i++) {
                    String[] nodeStr = nodeArr[i].split(":");
                    Address address = null;
                    if (nodeStr.length > 1) {
                        address = new Address(nodeStr[0], Integer.parseInt(nodeStr[1]));
                    } else {
                        address = new Address(nodeStr[0]);
                    }
                    addressList.add(address);
                }
                connection = factory.newConnection(addressList);
            } else {
                connection = factory.newConnection();
            }
            //创建一个频道
            channel = connection.createChannel();
        } catch (Exception e) {
            e.printStackTrace();
            instance = null;
            LOGGER.error("创建连接失败：" + e.getMessage(), e);
        }
    }

    public boolean sendMessage2Queue(String message, String queueName) {
        try {
            //指定一个队列
            channel.queueDeclare(queueName, true, false, false, null);
            //发送的消息
//            message = new String(message.getBytes("UTF-8"),"UTF-8");
            //往队列中发出一条消息
            channel.basicPublish("", queueName, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes(Charset.forName("UTF-8")));
            return true;
        } catch (Exception e) {
            LOGGER.error("发送消息失败：" + e.getMessage(), e);
            return false;
        }
    }

    /**
     * 发送广播消息
     *
     * @param exchange 交换器名称
     * @param message
     * @return
     */
    public boolean sendBroacastMsg2Exch(String exchange, String message) {
        try {
            channel.exchangeDeclare(exchange, "fanout", true, false, null);
            channel.basicPublish(exchange, "", MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes(Charset.forName("UTF-8")));
            return true;
        } catch (IOException e) {
            e.printStackTrace();
        }
        return false;
    }

    public void closeMq() {
        try {
            //关闭频道和连接
            channel.close();
            connection.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


}
